package com.ztorn.cdc;

import com.ztorn.cdc.doris.DorisSinkBuilder;
import com.ztorn.cdc.pulsar.PulsarSinkBuilder;
import com.ztorn.common.assertion.Asserts;
import com.ztorn.common.model.FlinkCDCConfig;

import java.util.HashMap;
import java.util.Map;
import java.util.function.Supplier;

public class SinkBuilderFactory {

    private static final Map<String, Supplier<SinkBuilder>> SINK_BUILDER_MAP =
            new HashMap<String, Supplier<SinkBuilder>>() {
                {
                    put(PulsarSinkBuilder.KEY_WORD, () -> new PulsarSinkBuilder());
                    put(DorisSinkBuilder.KEY_WORD, () -> new DorisSinkBuilder());
                }
            };

    public static SinkBuilder buildSinkBuilder(FlinkCDCConfig config) {
        if (Asserts.isNull(config) || Asserts.isNullString(config.getSink().get("connector"))) {
            throw new RuntimeException("请指定 Sink connector。");
        }
        return SINK_BUILDER_MAP
                .getOrDefault(config.getSink().get("connector"), () -> new PulsarSinkBuilder())
                .get()
                .create(config);
    }
}
